package mq

import (
	log "cyiris/core/zaplog"
	"errors"
	"fmt"
	"time"

	"github.com/streadway/amqp"
)

type ExchangeCfg struct {
	MqName       string `yaml:"mq_name"`
	ExchangeName string `yaml:"exchange_name"`
	ExchangeType string `yaml:"exchange_type"` // "direct", "fanout", "topic" and "headers"
}

type Publisher struct {
	*Channel
	ExchangeName string
	ExchangeType string // "direct", "fanout", "topic" and "headers"
}

func NewPublisher(conn *Connector, exchangeName, ExchangeType string) *Publisher {
	chann := NewChannel(conn)

	pub := &Publisher{
		Channel:      chann,
		ExchangeName: exchangeName,
		ExchangeType: ExchangeType,
	}

	pub.prepareExchange()

	return pub
}

//准备exchange
func (pub *Publisher) prepareExchange() error {
	err := pub.channel.ExchangeDeclare(pub.ExchangeName, pub.ExchangeType, true, false, false, false, nil)
	if err != nil {
		log.Errorf("Failed to declare a exchagne<%v>, err:%s", pub.ExchangeName, err)
		return err
	}

	return nil
}

/**
 * 发送消息到指定的 routingKey
 */
func (pub *Publisher) Push(routingKey string, data []byte) error {
	if !pub.isReady {
		return fmt.Errorf("failed to push: exchange<%v> channel not open", pub.ExchangeName)
	}

	for {
		err := pub.UnsafePush(routingKey, data)
		if err != nil {
			log.Warnf("exchange<%v> Push failed. Retrying...", pub.ExchangeName)
			select {
			case <-pub.done:
				return errors.New("session is shutting down")
			case <-time.After(resendDelay):
			}
			continue
		}

		select {
		case confirm := <-pub.notifyConfirm:
			if confirm.Ack {
				log.Infof("exchange<%v> Push confirmed!", pub.ExchangeName)
				return nil
			}
		case <-time.After(resendDelay):
		}

		log.Warnf("exchange<%v> Push didn't confirm. Retrying...", pub.ExchangeName)
	}
}

func (pub *Publisher) UnsafePush(routintKey string, data []byte) error {
	if !pub.isReady {
		return fmt.Errorf("failed to push: exchange<%v> channel not open", pub.ExchangeName)
	}
	return pub.channel.Publish(
		pub.ExchangeName, // Exchange
		routintKey,       // Routing key
		false,            // Mandatory
		false,            // Immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        data,
		},
	)
}
